package com.relytosoft.mizar.analyze.strategy.consumer;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.core.constant.CommonConstants;
import com.relytosoft.dubhe.common.core.object.ResponseResult;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.TaskRequestDto;
import com.relytosoft.mizar.analyze.strategy.algorithm.structure.service.StructureService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 实时结构化任务消费
 *
 * @Author SunLi
 * @Date: 2022/5/17 10:22
 * @Version 1.0
 */

@Component
public class StruTaskConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private StructureService structureService;

    @Resource
    private KafkaProducer kafkaProducer;

    @KafkaListener(id = "consumer-analyze-stru-task", topics = MediaMqConstants.STRU_ANALYZE_TASK_TOPIC, containerFactory = "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            boolean isConsume = false;
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            String data = kafkaOpMsg.getData();
            switch (kafkaOpMsg.getOpType()) {
                case "insert":
                    isConsume = createTask(data);
                    break;
                case "delete":
                    isConsume = stopTask(data);
                    break;
                default:
                    break;
            }
            log.info("线程:{} 消费-结构化任务->{}", Thread.currentThread(), data);
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 创建实时分析任务
     *
     * @param data 任务信息
     * @return 消费结果
     */
    private boolean createTask(String data) {
        TaskRequestDto taskRequestDto = JSON.parseObject(data, TaskRequestDto.class);
        ResponseResult result = structureService.createTask(taskRequestDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
            kafkaOpMsg.setData(JSON.toJSONString(result.getData()));
            kafkaProducer.send(MediaMqConstants.STRU_ANALYZE_TASK_ID_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }

    /**
     * 删除实时分析任务
     *
     * @param data 任务ID值
     * @return 消费结果
     */
    private boolean stopTask(String data) {
        ResponseResult result = structureService.stopTask(data);
        if (result.getCode() == CommonConstants.SUCCESS) {
            return true;
        }
        return false;
    }
}
